package cn.xinfei.xdecision.data.core.mq;

import cn.xinfei.xdecision.model.VariableMessageVO;
import cn.xinfei.xdecision.pulsar.XPulsarProducer;
import cn.xinfei.xdecision.pulsar.config.XDecisionMessage;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.MessageId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.UUID;
import java.util.concurrent.CompletableFuture;

@Slf4j
@Component
public class VariableProducer {


    @Autowired
    private XPulsarProducer pulsarProducer;
    @Value("${xdecision.data.variabletopic}")
    private String variableTopic;

    public void send(VariableMessageVO message, String requestId, String scene) {
        try {
            XDecisionMessage msg = new XDecisionMessage();
            msg.setData(JSONObject.toJSONString(message));
            msg.setKey(requestId + "_" + UUID.randomUUID().toString().substring(0, 8) + "_" + scene);
            msg.setMsgId(requestId + "_" + UUID.randomUUID().toString().substring(0, 8) + "_" + scene);
            CompletableFuture<MessageId> messageIdCompletableFuture = pulsarProducer.sendAsync(variableTopic, msg);
            messageIdCompletableFuture.thenAccept(msgId -> {
                log.info("Message with ID " + msgId + " successfully sent! mesg: " + JSONObject.toJSONString(msg));
            }).exceptionally(ex -> {
                log.error("request with ID " + requestId + " exception sent", ex);
                return null;
            });
        } catch (Exception exception) {
            exception.printStackTrace();
            log.error("error in variable mesg send! requestId = {} mesg = {} scene = {}",
                    requestId,
                    JSONObject.toJSONString(message),
                    scene);
        }
    }
}
